package com.hrpc.core;

import com.hrpc.NettyBootStrapInitializer;
import com.hrpc.compress.CompressorFactory;
import com.hrpc.discovery.Register;
import com.hrpc.enumeration.RequestType;
import com.hrpc.hRpcBootStarp;
import com.hrpc.serialize.SerializeFactory;
import com.hrpc.transport.message.hRpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 心跳探测的核心目的是什么？探活，感知哪些服务器的连接状态是正常的，哪些是不正常的
 */
public class HeartbeatDetector {
    //做心跳检测
    public static void detectHeart(String serviceName){
        //1.从注册中心拉取服务列表并建立连接
        Register register = hRpcBootStarp.getInstance().getRegister();
        List<InetSocketAddress> address = register.lookup(serviceName,hRpcBootStarp.getInstance().getConfiguration().getGroup());
        //将连接进行缓存
        for (InetSocketAddress inetSocketAddress : address) {
            try {
                if (!hRpcBootStarp.CHANNEL_CACHE.containsKey(inetSocketAddress)){
                    //获取连接
                    Channel channel = NettyBootStrapInitializer.getBootstrap().connect(inetSocketAddress).sync().channel();
                    hRpcBootStarp.CHANNEL_CACHE.put(inetSocketAddress,channel);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        //3.任务：定期发送消息
        Thread thread = new Thread(() ->
            new Timer().scheduleAtFixedRate(new MyTimeTask(), 5000, 5000)
        ,"hRpc-HeartbeatDetector-thread");
        //设置为守护线程，主线程没了，这个线程也就没了
        thread.setDaemon(true);
        thread.start();
    }
    private static class MyTimeTask extends TimerTask{

        @Override
        public void run() {
            //将响应时长的map清空
            hRpcBootStarp.ANSWER_TIME_CHANNEL_CACHE.clear();
            //遍历所有的channel
            Map<InetSocketAddress, Channel> channelCache = hRpcBootStarp.CHANNEL_CACHE;
            for (Map.Entry<InetSocketAddress,Channel> entry : channelCache.entrySet()){
                int tryTime=3;
                while (tryTime > 0){
                    Channel channel = entry.getValue();
                    long start=System.currentTimeMillis();
                    //构建一个心跳请求
                    hRpcRequest hRpcRequest=new hRpcRequest(
                            hRpcBootStarp.getInstance().getConfiguration().getId_generate().getID(),
                            RequestType.HEART_BEAT.getId(),
                            CompressorFactory.getCompressor(hRpcBootStarp.getInstance().getConfiguration().getCompressType()).getCode(),
                            SerializeFactory.getSerialize(hRpcBootStarp.getInstance().getConfiguration().getSerializeType()).getCode(),
                            start);
                    //4.写出报文
                    CompletableFuture<Object> completableFuture=new CompletableFuture<>();
                    // 需要将completableFuture暴露出去
                    hRpcBootStarp.PENDING_REQUEST.put(hRpcRequest.getRequestId(), completableFuture);
                    channel.writeAndFlush(hRpcRequest).addListener((ChannelFutureListener) promise->{
                        if (!promise.isSuccess()) {
                            completableFuture.completeExceptionally(promise.cause());
                        }
                    });
                    Long endTime = 0L;
                    try {
                        completableFuture.get(1, TimeUnit.SECONDS);
                        endTime = System.currentTimeMillis();
                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
                        tryTime--;
                        //失效的地址移除我们的服务列表
                        if (tryTime == 0){
                            hRpcBootStarp.CHANNEL_CACHE.remove(entry.getKey());
                        }
                        try {
                            Thread.sleep(10*new Random().nextInt(5));
                        } catch (InterruptedException ex) {
                            throw new RuntimeException(ex);
                        }
                        continue;
                    }
                    Long time = endTime - start;
                    //使用treemap（有序，找最大最小值很快）进行缓存
                    hRpcBootStarp.ANSWER_TIME_CHANNEL_CACHE.put(time,channel);
                    System.out.println(time);
                }
            }
        }
    }
}
